package com.atcyj.gmall.realtime.dwd.db.app;

import com.atcyj.gamll.realtime.common.base.BaseSqlApp;
import com.atcyj.gamll.realtime.common.constant.Constant;
import com.atcyj.gamll.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author: cyj
 * @date: 2024/8/18
 */
public class DwdTradeOrderCancelDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeOrderCancelDetail().start(
                10014,
                4,
                "dwd_trade_order_cancel_detail"
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        // 两个打标进行关联，用 regular join， 需要设置ttl状态存活时间
        // 电商平台上，当提交了订单后，如果不在指定的时间范围（15min）内付款，就会默认取消订单，或者在这时间范围内，主动取消订单，
        // 所以当取消订单数据还没来之前，下单事务事实表的数据需要在状态中保留至少15min+5s
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(15 * 60 + 5L));


        // 1. 从交易域下单事务事实表主题中读取数据，创建对应的动态表
        tableEnv.executeSql("create table dwd_trade_order_detail(\n" +
                "      id string ,\n" +
                "      order_id string ,\n" +
                "      user_id string ,\n" +
                "      province_id string ,\n" +
                "      sku_id string ,\n" +
                "      sku_name string ,\n" +
                "      split_original_amount string ,\n" +
                "      sku_num string ,\n" +
                "      activity_id string ,\n" +
                "      activity_rule_id string ,\n" +
                "      coupon_id string ,\n" +
                "      create_time string ,\n" +
                "      split_total_amount string ,\n" +
                "      split_activity_amount string ,\n" +
                "      split_coupon_amount string \n" +
                ")" + SQLUtil.getKafkaSourceSQL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, "dwd_trade_order_cancel_detail"));

        // 2. 从topic_db中读取取消订单的数据

        // 读取topic_db数据
        createTopicDB("dwd_trade_order_cancel_detail", tableEnv);


        Table orderCancelTable = tableEnv.sqlQuery(
                "select\n" +
                        "  `data`['id'] id,\n" +
                        "  `data`['operate_time'] operate_time,\n" +
                        "  ts \n" +
                        "from topic_db\n" +
                        "where `database`='gmall'\n" +
                        "  and `table`= 'order_info'\n" +
                        "  and `type`='update'\n" +
                        "  and `data`['order_status']='1003'\n" +
                        "  and  `old`['order_status']='1001'   \n");
        tableEnv.createTemporaryView("order_cancel", orderCancelTable);

        // 3. 两表做关联
        Table result = tableEnv.sqlQuery(
                "select\n" +
                        "  od.id,\n" +
                        "  od.order_id,\n" +
                        "  od.user_id,\n" +
                        "  od.province_id,\n" +
                        "  od.sku_id,\n" +
                        "  od.sku_name,\n" +
                        "  od.split_original_amount,\n" +
                        "  od.sku_num,\n" +
                        "  od.activity_id,\n" +
                        "  od.activity_rule_id,\n" +
                        "  od.coupon_id,\n" +
                        "  date_format(oc.operate_time,'yyyy-MM-dd') order_cancel_date,\n" +
                        "  oc.operate_time cancel_time,\n" +
                        "  od.split_total_amount,\n" +
                        "  od.split_activity_amount,\n" +
                        "  od.split_coupon_amount,\n" +
                        "  oc.ts \n" +
                        "from dwd_trade_order_detail od \n" +
                        "inner join order_cancel oc on od.order_id = oc.id \n");


        // 4. 创建取消订单主题对应的动态表
        tableEnv.executeSql(
                "create table  " + Constant.TOPIC_DWD_TRADE_ORDER_CANCEL + "(\n" +
                        "  id string ,\n" +
                        "  order_id string ,\n" +
                        "  user_id string ,\n" +
                        "  province_id string ,\n" +
                        "  sku_id string ,\n" +
                        "  sku_name string ,\n" +
                        "  split_original_amount string ,\n" +
                        "  sku_num string ,\n" +
                        "  activity_id string ,\n" +
                        "  activity_rule_id string ,\n" +
                        "  coupon_id string ,\n" +
                        "  order_cancel_date string, \n" +
                        "  cancel_time string  ,\n" +
                        "  split_total_amount string ,\n" +
                        "  split_activity_amount string ,\n" +
                        "  split_coupon_amount string ,\n" +
                        "  ts bigint,\n" +
                        "  PRIMARY KEY (id) NOT ENFORCED  \n" +
                        ")" + SQLUtil.getKafkaUpsertSinkSQL(Constant.TOPIC_DWD_TRADE_ORDER_CANCEL));

        // 5. 数据插入到取消订单动态表中，完成写出到kafka主题的操作
        result.insertInto(Constant.TOPIC_DWD_TRADE_ORDER_CANCEL).execute();
    }
}
